package realtime.app.dws;/*
 *@ClassName : Dws_Trade_Source_TradeSum
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-27 14:{MINUTE}
 */

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TradeSourceBean;
import realtime.func.AsyncDIMFunction;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class Dws_Trade_Source_TradeSum {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_source_tradeSum_0327_08";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    if (s != null) {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        //if (!(jsonObject.getString("source_id" )==null )){
                            collector.collect(jsonObject);
                        //}
                    }
                } catch (JSONException e) {
                    System.out.println(s + " : 无法转换为json格式");
                }
            }
        });

        //
        SingleOutputStreamOperator<JSONObject> jsonWithWMDS = jsonDS
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("create_time");
                            }
                        }));

        //
        SingleOutputStreamOperator<TradeSourceBean> beanDS = jsonWithWMDS.map(new MapFunction<JSONObject, TradeSourceBean>() {
            @Override
            public TradeSourceBean map(JSONObject jsonObject) throws Exception {

                //System.out.println("jsonObject>>>" + jsonObject);

                String orderId = jsonObject.getString("order_id");
                String userId = jsonObject.getString("user_id");
                String sourceId = jsonObject.getString("source_id");

                //System.out.println("sourceId>>>>>" + sourceId);

                return TradeSourceBean.builder()
                        .sourceId(sourceId)
                        .finalAmount(jsonObject.getDouble("final_amount"))
                        .orderSet(new HashSet<String>(Collections.singleton(orderId)))
                        .userSet(new HashSet<String>(Collections.singleton(userId)))
                        .build();
            }
        });

        //beanDS.print("keyed>>>>>>>>>>>>>>>>>>\n");已通过

        //
        KeyedStream<TradeSourceBean, String> keyedStream = beanDS.keyBy(bean -> bean.getSourceId());

        //keyedStream.print("keyed>>>>>>>>>>>>>>>>>>\n");已通过

        //
        SingleOutputStreamOperator<TradeSourceBean> reduceDS = keyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TradeSourceBean>() {

                    @Override
                    public TradeSourceBean reduce(TradeSourceBean t1, TradeSourceBean t2) throws Exception {

                        t1.setFinalAmount(t1.getFinalAmount() + t2.getFinalAmount());
                        t1.getUserSet().addAll(t2.getUserSet());
                        t1.getOrderSet().addAll(t2.getOrderSet());

                        return t1;
                    }
                }, new WindowFunction<TradeSourceBean, TradeSourceBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeSourceBean> iterable, Collector<TradeSourceBean> collector) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());

                        for (TradeSourceBean bean : iterable) {
                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setPersonCount((long) (bean.getUserSet().size()));
                            bean.setOrderCount((long) (bean.getOrderSet().size()));
                            bean.setTs(System.currentTimeMillis());

                            collector.collect(bean);
                        }
                    }
                });

        //reduceDS.print("keyedStream>>>>>>>>>>>>>>>>\n");

        //
        SingleOutputStreamOperator<TradeSourceBean> resultDS = AsyncDataStream.unorderedWait(
                reduceDS
                , new AsyncDIMFunction<TradeSourceBean>("DIM_BASE_SOURCE") {

                    @Override
                    public void join(TradeSourceBean input, JSONObject dimInfo) {
                        input.setSourceName(dimInfo.getString("SOURCE_SITE"));
                    }

                    @Override
                    public String getKey(TradeSourceBean input) {
                        return input.getSourceId();
                    }
                }
                , 60
                , TimeUnit.SECONDS
        );

        //
        resultDS.print("Dws_Trade_Source_TradeSum>>>>>>>>>>>>>>>>>>>>>>>\n");
        resultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_source_tradesum_window values(?,?,?,?,?,?,?,?)"));

        //
        env.execute();


    }
}
